package com.zsd.android.rxjavalib;

import android.os.Handler;
import android.os.Looper;

public class ObserverOn<T> implements ObservableOnSubscribe<T> {

    private ObservableOnSubscribe<T> source;

    public ObserverOn(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    public void subscribe(Observer<? super T> observableEmitter) {
        PackageObserver<T> packageObserver = new PackageObserver(observableEmitter);
        source.subscribe(packageObserver);
    }

    private final class PackageObserver<T> implements Observer<T> {

        private Observer<T> observer;
        private Handler mHandler;

        public PackageObserver(Observer<T> observer) {
            this.observer = observer;
            mHandler = new Handler(Looper.getMainLooper());
        }

        @Override
        public void onSubscribe() {

        }

        @Override
        public void onNext(final T item) {
            mHandler.post(new Runnable() {
                @Override
                public void run() {
                    observer.onNext(item);
                }
            });
        }

        @Override
        public void onError(final Throwable e) {
            mHandler.post(new Runnable() {
                @Override
                public void run() {
                    observer.onError(e);
                }
            });
        }

        @Override
        public void onComplete() {
            mHandler.post(new Runnable() {
                @Override
                public void run() {
                    observer.onComplete();
                }
            });
        }
    }
}
